package org.gxlu.com.jstorm.day1;

import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

public class CountBolt  extends BaseRichBolt {
	private static Logger log = LoggerFactory.getLogger(CountBolt.class) ;
	
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;
	private OutputCollector collector = null ;
	
	Map<String,Integer> map = new HashMap<String,Integer>();
	
	@Override
	public void execute(Tuple tuple) {
		 String key = tuple.getString(0);
		 Integer num = tuple.getInteger(1);
		 
		 if(map.containsKey(key)){
			 Integer count = map.get(key);
			 map.put(key, num+count) ;
		 }else{
			 map.put(key, num) ;
		 }
		 log.info("CountBolt-----threadId:"+Thread.currentThread().getId()+"map.hashcode:"+map.getClass().hashCode()+",map:"+map);
		    
        // Utils.sleep(3000);
		// System.out.println("str---:"+str);
	}

	@Override
	public void prepare(Map arg0, TopologyContext arg1, OutputCollector arg2) {
		 System.out.println("---StringWopriBolt--prepare-") ;
		 collector = arg2 ;

	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declare) {
		 //declare.declare(new Fields("world"));

	}

}